package cn.lsh.hbase;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * Hbase封装服务
 *
 * @author liush
 * @date 2020/8/31 10:29
 */
public class HbaseServiceImpl {
	Connection conn;

	public HbaseServiceImpl() {
		Configuration conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.property.clientPort", "2181");
		conf.set("hbase.zookeeper.quorum", "node01,node02,node03");
		try {
			conn = ConnectionFactory.createConnection(conf);
		} catch (IOException e) {
			throw new RuntimeException(e);
		}
	}

	/**
	 * 构建admin
	 *
	 * @return HBaseAdmin
	 * @throws Exception 抛出可能的连接异常
	 */
	public HBaseAdmin buildAdmin() throws Exception {
		return (HBaseAdmin) conn.getAdmin();
	}

	/**
	 * 构建table
	 *
	 * @param tableName hbase表名
	 * @return 表的操作对象HTable
	 * @throws Exception 抛出可能的连接异常
	 */
	public HTable buildHTable(String tableName) throws Exception {
		return (HTable) conn.getTable(TableName.valueOf(tableName));
	}

	/**
	 * 关闭hbase连接
	 */
	public void closeConnect() {
		try {
			if (conn != null) {
				conn.close();
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public void closeAdmin(Admin admin) {
		try {
			if (admin != null) {
				admin.close();
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public void closeHTable(HTable hTable) {
		try {
			if (hTable != null) {
				hTable.close();
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 判断表是否存在
	 *
	 * @param tableName hbase表名
	 * @return boolean
	 * @throws Exception 抛出可能的连接异常
	 */
	public boolean tableExists(String tableName) throws Exception {
		TableName tn = TableName.valueOf(tableName);
		HBaseAdmin admin = buildAdmin();
		return admin.tableExists(tn);
	}

	/**
	 * 创建表
	 *
	 * @param tableName    hbase表名
	 * @param columnFamily 列族
	 * @return boolean
	 */
	public boolean createTable(String tableName, String columnFamily) {
		HBaseAdmin admin = null;
		try {
			TableName tn = TableName.valueOf(tableName);
			admin = buildAdmin();
			if (admin.tableExists(tn)) {
				return false;
			}
			TableDescriptor desc = TableDescriptorBuilder.newBuilder(tn)
					.setColumnFamily(ColumnFamilyDescriptorBuilder.of(columnFamily)).build();
			admin.createTable(desc);
			return true;
		} catch (Exception e) {
			e.printStackTrace();
			return false;
		} finally {
			closeAdmin(admin);
		}
	}

	/**
	 * 删除表
	 *
	 * @param tableName hbase表名
	 * @return boolean
	 */
	public boolean deleteTable(String tableName) {
		HBaseAdmin admin = null;
		try {
			admin = buildAdmin();
			TableName tn = TableName.valueOf(tableName);
			if (admin.tableExists(tn)) {
				admin.disableTable(tn);
				admin.deleteTable(tn);
			}
			return true;
		} catch (Exception e) {
			e.printStackTrace();
			return false;
		} finally {
			closeAdmin(admin);
		}
	}

	/**
	 * 添加列族
	 *
	 * @param tableName     hbase表名
	 * @param columnFamilys 多个列族，可变参数
	 * @return boolean
	 */
	public boolean addColumnFamily(String tableName, String... columnFamilys) {
		HBaseAdmin admin = null;
		try {
			TableName tn = TableName.valueOf(tableName);
			admin = buildAdmin();
			if (!admin.tableExists(tn)) {
				return true;
			}
			for (String cf : columnFamilys) {
				admin.addColumnFamily(tn, ColumnFamilyDescriptorBuilder.of(cf));
			}
			return true;
		} catch (Exception e) {
			e.printStackTrace();
			return false;
		} finally {
			closeAdmin(admin);
		}
	}

	/**
	 * 删除列族
	 *
	 * @param tableName     hbase表名
	 * @param columnFamilys 列族
	 * @return boolean
	 */
	public boolean deleteColumnFamily(String tableName, String... columnFamilys) {
		HBaseAdmin admin = null;
		try {
			TableName tn = TableName.valueOf(tableName);
			admin = buildAdmin();
			if (!admin.tableExists(tn)) {
				return false;
			}
			for (String cf : columnFamilys) {
				admin.deleteColumnFamily(tn, Bytes.toBytes(cf));
			}
			return true;
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			closeAdmin(admin);
		}
		return false;
	}

	public boolean insertRow(String tableName, Put put) {
		HTable hTable = null;
		try {
			hTable = buildHTable(tableName);
			hTable.put(put);
			return true;
		} catch (Exception e) {
			e.printStackTrace();
			return false;
		} finally {
			closeHTable(hTable);
		}
	}

	public boolean insertRow(String tableName, List<Put> puts) {
		HTable hTable = null;
		try {
			hTable = buildHTable(tableName);
			hTable.put(puts);
			return true;
		} catch (Exception e) {
			e.printStackTrace();
			return false;
		} finally {
			closeHTable(hTable);
		}
	}

	/**
	 * 一个列族下插入一列
	 *
	 * @param tableName    hbase表名
	 * @param rowKey       行标识
	 * @param columnFamily 列族
	 * @param column       列名
	 * @param value        列值
	 * @return boolean
	 */
	public boolean insertRowOneColumn(String tableName, String rowKey, String columnFamily, String column, String value) {
		Put put = new Put(Bytes.toBytes(rowKey));
		put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
		return insertRow(tableName, put);
	}

	/**
	 * 一个列族下插入多列
	 *
	 * @param tableName    hbase表名
	 * @param rowKey       行标识
	 * @param columnFamily 列族
	 * @param columns      列名
	 * @param values       列值
	 * @return boolean
	 */
	public boolean insertRowMoreColumn(String tableName, String rowKey, String columnFamily, String[] columns, Object[] values) {
		if (columns.length != values.length) {
			return false;
		}
		Put put = new Put(Bytes.toBytes(rowKey));
		for (int i = 0; i < columns.length; i++) {
			put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columns[i]), tran(values[i]));
		}
		return insertRow(tableName, put);
	}

	public boolean insert(String tableName, List<Put> puts) {
		HTable hTable = null;
		try {
			hTable = buildHTable(tableName);
			hTable.put(puts);
			return true;
		} catch (Exception e) {
			e.printStackTrace();
			return false;
		} finally {
			closeHTable(hTable);
		}
	}

	/**
	 * 获取一整行数据
	 *
	 * @param tableName hbase表名
	 * @param rowKey    行标识
	 * @return Result
	 */
	public Result getOneRow(String tableName, String rowKey) {
		HTable hTable = null;
		try {
			hTable = buildHTable(tableName);
			Get get = new Get(Bytes.toBytes(rowKey));
			return hTable.get(get);
		} catch (Exception e) {
			e.printStackTrace();
			return null;
		} finally {
			closeHTable(hTable);
		}
	}

	/**
	 * 查询一行的多列数据
	 *
	 * @param tableName    hbase表名
	 * @param rowKey       行标识
	 * @param columnFamily 列族
	 * @param cols         列名
	 * @return Result
	 */
	public Result getOneRowAndMultiColumn(String tableName, String rowKey, String columnFamily, String[] cols) {
		HTable hTable = null;
		try {
			hTable = buildHTable(tableName);
			Get get = new Get(Bytes.toBytes(rowKey));
			if (cols != null && cols.length > 0) {
				for (String col : cols) {
					get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(col));
				}
			} else {
				get.addFamily(Bytes.toBytes(columnFamily));
			}
			return hTable.get(get);
		} catch (Exception e) {
			e.printStackTrace();
			return null;
		} finally {
			closeHTable(hTable);
		}
	}

	/**
	 * 根据过滤条件查询数据
	 *
	 * @param tableName    hbase表名
	 * @param rowKeyPrefix 行标识前置过滤
	 * @param filters      其他过滤
	 * @return Result列表
	 */
	public List<Result> getRows(String tableName, String rowKeyPrefix, Filter... filters) {
		FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
		Filter preFilter = new PrefixFilter(Bytes.toBytes(rowKeyPrefix));
		filterList.addFilter(preFilter);
		if (filters != null && filters.length > 0) {
			for (Filter filter : filters) {
				filterList.addFilter(filter);
			}
		}
		Scan scan = new Scan();
		scan.setFilter(filterList);
		return getRows(tableName, scan);
	}

	/**
	 * 范围查询
	 *
	 * @param tableName   hbase表名
	 * @param startRowKey 起始位置
	 * @param stopRowKey  结束位置
	 * @return Result列表
	 */
	public List<Result> getRows(String tableName, String startRowKey, String stopRowKey) {
		Scan scan = new Scan();
		scan.withStartRow(Bytes.toBytes(startRowKey));
		scan.withStopRow(Bytes.toBytes(stopRowKey));
		return getRows(tableName, scan);
	}

	/**
	 * 获取指定rowKey前缀的指定列
	 *
	 * @param tableName    hbase表名
	 * @param rowKeyPrefix 行标识前置过滤
	 * @param columnFamily 列族
	 * @param cols         列名
	 * @return Result列表
	 */
	public List<Result> getRowsByColumn(String tableName, String rowKeyPrefix, String columnFamily, String[] cols) {
		Scan scan = new Scan();
		scan.setFilter(new PrefixFilter(Bytes.toBytes(rowKeyPrefix)));
		if (cols != null && cols.length > 0) {
			for (String col : cols) {
				scan.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(col));
			}
		} else if (StringUtils.isNotBlank(columnFamily)) {
			scan.addFamily(Bytes.toBytes(columnFamily));
		}
		return getRows(tableName, scan);
	}

	public List<Result> getRows(String tableName, Scan scan) {
		HTable hTable = null;
		ResultScanner rs = null;
		List<Result> resultList = new ArrayList<>();
		try {
			hTable = buildHTable(tableName);
			rs = hTable.getScanner(scan);
			for (Result result : rs) {
				resultList.add(result);
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if (rs != null) {
				rs.close();
			}
			closeHTable(hTable);
		}
		return resultList;
	}

	/**
	 * 获取一整行数据
	 *
	 * @param tableName hbase表名
	 * @param rowKey    行标识
	 * @return Map 映射类型 columnFamily：column：value
	 */
	public Map<String, Map<String, String>> getOneRowOfMap(String tableName, String rowKey) {
		Result result = getOneRow(tableName, rowKey);
		return HbaseUtil.resultToMap(result, String.class);
	}

	public ResultScanner scan(String tableName) {
		return scan(tableName, new Scan());
	}

	public ResultScanner scan(String tableName, String columnFamily) {
		Scan scan = new Scan();
		scan.addFamily(Bytes.toBytes(columnFamily));
		return scan(tableName, new Scan());
	}

	public ResultScanner scan(String tableName, String columnFamily, String column) {
		Scan scan = new Scan();
		scan.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
		return scan(tableName, new Scan());
	}

	public ResultScanner scan(String tableName, Scan scan) {
		HTable hTable = null;
		try {
			hTable = buildHTable(tableName);
			return hTable.getScanner(scan);
		} catch (Exception e) {
			e.printStackTrace();
			return null;
		} finally {
			closeHTable(hTable);
		}
	}

	/**
	 * 删除多个单元格，即删除多列
	 *
	 * @param tableName    hbase表名
	 * @param rowKey       行标识
	 * @param columnFamily 列族
	 * @param columns      列名
	 * @return 删除记录数
	 */
	public int deleteCells(String tableName, String rowKey, String columnFamily, String[] columns) {
		if (StringUtils.isBlank(rowKey)) {
			return 0;
		}
		HTable hTable = null;
		try {
			hTable = buildHTable(tableName);
			if (columns != null && columns.length > 0) {
				List<Delete> deletes = new ArrayList<>();
				for (String column : columns) {
					Delete delete = new Delete(Bytes.toBytes(rowKey));
					delete.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
					deletes.add(delete);
				}
				hTable.delete(deletes);
				return columns.length;
			}
			return 0;
		} catch (Exception e) {
			e.printStackTrace();
			return 0;
		} finally {
			closeHTable(hTable);
		}
	}

	/**
	 * 删除列族
	 *
	 * @param tableName    hbase表名
	 * @param rowKey       行标识
	 * @param columnFamily 列族
	 * @return 删除记录数
	 */
	public int deleteColumnFamily(String tableName, String rowKey, String columnFamily) {
		if (StringUtils.isBlank(rowKey) || StringUtils.isBlank(columnFamily)) {
			return 0;
		}
		HTable hTable = null;
		try {
			hTable = buildHTable(tableName);
			Delete delete = new Delete(Bytes.toBytes(rowKey));
			delete.addFamily(Bytes.toBytes(columnFamily));
			hTable.delete(delete);
			return 1;
		} catch (Exception e) {
			e.printStackTrace();
			return 0;
		} finally {
			closeHTable(hTable);
		}
	}

	/**
	 * 删除一行数据
	 *
	 * @param tableName hbase表名
	 * @param rowKey    行标识
	 * @return 删除记录数
	 */
	public int deleteRecord(String tableName, String rowKey) {
		if (StringUtils.isBlank(rowKey)) {
			return 0;
		}
		HTable hTable = null;
		try {
			hTable = buildHTable(tableName);
			Delete delete = new Delete(Bytes.toBytes(rowKey));
			hTable.delete(delete);
			return 1;
		} catch (Exception e) {
			e.printStackTrace();
			return 0;
		} finally {
			closeHTable(hTable);
		}
	}

	/**
	 * 根据前缀删除n行
	 *
	 * @param tableName    hbase表名
	 * @param rowKeyPrefix 行标识前置过滤
	 * @return 删除行数
	 */
	public int deleteRecords(String tableName, String rowKeyPrefix) {
		Scan scan = new Scan();
		scan.setFilter(new PrefixFilter(Bytes.toBytes(rowKeyPrefix)));
		return deleteRecords(tableName, scan);
	}

	/**
	 * 删除表的所有列
	 *
	 * @param tableName hbase表名
	 * @return 删除记录数
	 */
	public int deleteTableData(String tableName) {
		return deleteRecords(tableName, new Scan());
	}

	/**
	 * 删除指定表的指定列族的指定列
	 *
	 * @param tableName    hbase表名
	 * @param columnFamily 列族
	 * @param cols         列名
	 * @return 删除记录数
	 */
	public int deleteTableData(String tableName, String columnFamily, String[] cols) {
		Scan scan = new Scan();
		if (cols != null && cols.length > 0) {
			for (String col : cols) {
				scan.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(col));
			}
		} else if (StringUtils.isNotBlank(columnFamily)) {
			scan.addFamily(Bytes.toBytes(columnFamily));
		}
		return deleteRecords(tableName, scan, true);
	}

	public int deleteRecords(String tableName, Scan scan) {
		return deleteRecords(tableName, scan, false);
	}

	/**
	 * 根据扫描结果删除记录
	 *
	 * @param tableName
	 * @param scan
	 * @param filterCol 是否过滤列，即只删除行中的某列时，该值需要设置为true
	 * @return
	 */
	public int deleteRecords(String tableName, Scan scan, boolean filterCol) {
		HTable hTable = null;
		ResultScanner rs = null;
		try {
			hTable = buildHTable(tableName);
			rs = hTable.getScanner(scan);
			List<Delete> deletes = new ArrayList<>();
			for (Result result : rs) {
				Delete delete = new Delete(result.getRow());
				if (filterCol) {
					List<Cell> cells = result.listCells();
					for (Cell c : cells) {
						delete.addColumn(CellUtil.cloneFamily(c), CellUtil.cloneQualifier(c));
					}
				}
				deletes.add(delete);
			}
			int count = deletes.size();
			hTable.delete(deletes);
			return count;
		} catch (Exception e) {
			e.printStackTrace();
			return 0;
		} finally {
			if (rs != null) {
				rs.close();
			}
			closeHTable(hTable);
		}
	}

	/**
	 * 开启表的默认协处理器
	 *
	 * @param tableName
	 * @return
	 */
	public boolean enableAggregateService(String tableName) {
		return enableAggregateService(tableName, "org.apache.hadoop.hbase.coprocessor.AggregateImplementation");
	}

	public boolean enableAggregateService(String tableName, String coprocessor) {
		return modifyAggregateService(tableName, coprocessor, true);
	}

	public boolean disableAggregateService(String tableName) {
		return modifyAggregateService(tableName, "org.apache.hadoop.hbase.coprocessor.AggregateImplementation", false);
	}

	public boolean disableAggregateService(String tableName, String coprocessor) {
		return modifyAggregateService(tableName, coprocessor, false);
	}

	/**
	 * 添加删除表的协处理器
	 *
	 * @param tableName
	 * @param coprocessor 协处理器的类全路径
	 * @param isAdd       true为添加，false为删除
	 * @return
	 */
	private boolean modifyAggregateService(String tableName, String coprocessor, boolean isAdd) {
		Admin admin = null;
		HTable hTable = null;
		TableName tn = TableName.valueOf(tableName);
		try {
			admin = buildAdmin();
			if (!admin.tableExists(tn)) {
				return false;
			}
			hTable = buildHTable(tableName);
			TableDescriptor oldTableDescriptor = hTable.getDescriptor();
			TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(oldTableDescriptor);
			if (isAdd) {
				//添加协处理器
				builder = builder.setCoprocessor(coprocessor);
			} else {
				//移除协处理器
				builder = builder.removeCoprocessor(coprocessor);
			}
			admin.disableTable(tn);
			admin.modifyTable(builder.build());
			admin.enableTable(tn);
			return true;
		} catch (Exception e) {
			try {
				if (admin != null && admin.isTableDisabled(tn)) {
					admin.enableTable(tn);
				}
			} catch (IOException ioException) {
				ioException.printStackTrace();
			}
			e.printStackTrace();
		} finally {
			closeHTable(hTable);
			closeAdmin(admin);
		}
		return false;
	}

	private byte[] tran(Object o) {
		byte[] buff = null;
		if (o instanceof String) {
			buff = Bytes.toBytes((String) o);
		} else if (o instanceof Integer) {
			buff = Bytes.toBytes((Integer) o);
		} else if (o instanceof Long) {
			buff = Bytes.toBytes((Long) o);
		} else if (o instanceof Double) {
			buff = Bytes.toBytes((Double) o);
		} else if (o instanceof Float) {
			buff = Bytes.toBytes((Float) o);
		} else if (o instanceof Short) {
			buff = Bytes.toBytes((Short) o);
		} else if (o instanceof Boolean) {
			buff = Bytes.toBytes((Boolean) o);
		} else if (o instanceof BigDecimal) {
			buff = Bytes.toBytes((BigDecimal) o);
		} else if (o instanceof ByteBuffer) {
			buff = Bytes.toBytes((ByteBuffer) o);
		}
		return buff;
	}
}
